package com.kgc.bigdata.spark.sql

import org.apache.spark.{SparkConf, SparkContext}

/**
  * DataFrame常用操作案例
  */
object DataFrameApp {

  def main(args: Array[String]) {
    val sparkConf = new SparkConf().setMaster("local[2]").setAppName("DataFrameApp")
    val sc = new SparkContext(sparkConf)
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)

    /** 使用SQLContext将JSON文件转成DataFrame
      * people.json内容如下
      * {"name":"Michael"}
      * {"name":"Andy", "age":30}
      * {"name":"Justin", "age":19}
      */
    val df = sqlContext.read.json("H:/workspace/SparkProject/src/data/people.json")

    // 使用show方法将DataFrame的内容输出
    df.show

    /**
      * 运行结果如下
      * +----+-------+
      * | age|   name|
      * +----+-------+
      * |null|Michael|
      * |  30|   Andy|
      * |  19| Justin|
      * +----+-------+
      */

    // 使用printSchema方法输出DataFrame的Schema信息
    df.printSchema()

    /**
      * 运行结果如下
      * root
      * |-- age: long (nullable = true)
      * |-- name: string (nullable = true)
      */


    // 使用select方法来选择我们所需要的字段
    df.select("name").show()

    /**
      * 运行结果如下
      * +-------+
      * |   name|
      * +-------+
      * |Michael|
      * |   Andy|
      * | Justin|
      * +-------+
      */

    // 使用select方法选择我们所需要的字段，并未age字段加1
    df.select(df("name"), df("age") + 1).show()

    /**
      * 运行结果如下
      * +-------+---------+
      * |   name|(age + 1)|
      * +-------+---------+
      * |Michael|     null|
      * |   Andy|       31|
      * | Justin|       20|
      * +-------+---------+
      */

    //  使用filter方法完成条件过滤
    df.filter(df("age") > 21).show()

    /**
      * 运行结果如下
      * +---+----+
      * |age|name|
      * +---+----+
      * | 30|Andy|
      * +---+----+
      */

    // 使用groupBy方法进行分组，求分组后的总数
    df.groupBy("age").count().show()

    /**
      * 运行结果如下
      * +----+-----+
      * | age|count|
      * +----+-----+
      * |null|    1|
      * |  19|    1|
      * |  30|    1|
      * +----+-----+
      */

    //sql()方法执行SQL查询操作
    sqlContext.sql("SELECT * FROM table_name").show

    sc.stop
  }
}
